/*-------------------------------------------------------------------------
 *
 * toasting.c
 *	  此文件包含支持创建toast表的常规例程
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/catalog/toasting.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/genam.h"
#include "access/heapam.h"
#include "access/toast_compression.h"
#include "access/xact.h"
#include "catalog/binary_upgrade.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/heap.h"
#include "catalog/index.h"
#include "catalog/namespace.h"
#include "catalog/pg_am.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_opclass.h"
#include "catalog/pg_type.h"
#include "catalog/toasting.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "storage/lock.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/rel.h"
#include "utils/syscache.h"

static void fc_CheckAndCreateToastTable(Oid fc_relOid, Datum fc_reloptions,
									 LOCKMODE fc_lockmode, bool fc_check,
									 Oid fc_OIDOldToast);
static bool fc_create_toast_table(Relation fc_rel, Oid fc_toastOid, Oid fc_toastIndexOid,
							   Datum fc_reloptions, LOCKMODE fc_lockmode, bool fc_check,
							   Oid fc_OIDOldToast);
static bool fc_needs_toast_table(Relation fc_rel);


/*
 * CreateToastTable 变体
 *		如果表需要一个 toast 表，并且尚未拥有，
 *		则为其创建一个 toast 表。
 *
 * toast 表的 reloptions 也可以传递。传递 (Datum) 0
 * 以使用默认的 reloptions。
 *
 * 我们预计调用者已经验证关系是一个表，并已
 * 完成任何必要的权限检查。调用者期望此函数
 * 在进行任何更改时以 CommandCounterIncrement 结束。
 */
void AlterTableCreateToastTable(Oid fc_relOid, Datum fc_reloptions, LOCKMODE fc_lockmode)
{
	fc_CheckAndCreateToastTable(fc_relOid, fc_reloptions, fc_lockmode, true, InvalidOid);
}

void NewHeapCreateToastTable(Oid fc_relOid, Datum fc_reloptions, LOCKMODE fc_lockmode,
						Oid fc_OIDOldToast)
{
	fc_CheckAndCreateToastTable(fc_relOid, fc_reloptions, fc_lockmode, false, fc_OIDOldToast);
}

void NewRelationCreateToastTable(Oid fc_relOid, Datum fc_reloptions)
{
	fc_CheckAndCreateToastTable(fc_relOid, fc_reloptions, AccessExclusiveLock, false,
							 InvalidOid);
}

static void fc_CheckAndCreateToastTable(Oid fc_relOid, Datum fc_reloptions, LOCKMODE fc_lockmode,
						 bool fc_check, Oid fc_OIDOldToast)
{
	Relation	fc_rel;

	fc_rel = table_open(fc_relOid, fc_lockmode);

	/* create_toast_table 完成所有工作 */
	(void) fc_create_toast_table(fc_rel, InvalidOid, InvalidOid, fc_reloptions, fc_lockmode,
							  fc_check, fc_OIDOldToast);

	table_close(fc_rel, NoLock);
}

/*
 * 在引导期间创建一个 toast 表
 *
 * 在这里我们需要预先指定 toast 表及其索引的 OID
 */
void BootstrapToastTable(char *fc_relName, Oid fc_toastOid, Oid fc_toastIndexOid)
{
	Relation	fc_rel;

	fc_rel = table_openrv(makeRangeVar(NULL, fc_relName, -1), AccessExclusiveLock);

	if (fc_rel->rd_rel->relkind != RELKIND_RELATION &&
		fc_rel->rd_rel->relkind != RELKIND_MATVIEW)
		elog(ERROR, "\"%s\" is not a table or materialized view",
			 fc_relName);

	/* create_toast_table 完成所有工作 */
	if (!fc_create_toast_table(fc_rel, fc_toastOid, fc_toastIndexOid, (Datum) 0,
							AccessExclusiveLock, false, InvalidOid))
		elog(ERROR, "\"%s\" does not require a toast table",
			 fc_relName);

	table_close(fc_rel, NoLock);
}


/*
 * create_toast_table --- 内部工作马
 *
 * rel 已经打开并锁定
 * toastOid 和 toastIndexOid 通常是 InvalidOid，但在
 * 引导期间，它们可以是非零的，以指定手动分配的 OID
 */
static bool fc_create_toast_table(Relation fc_rel, Oid fc_toastOid, Oid fc_toastIndexOid,
				   Datum fc_reloptions, LOCKMODE fc_lockmode, bool fc_check,
				   Oid fc_OIDOldToast)
{
	Oid			fc_relOid = RelationGetRelid(fc_rel);
	HeapTuple	fc_reltup;
	TupleDesc	fc_tupdesc;
	bool		fc_shared_relation;
	bool		fc_mapped_relation;
	Relation	fc_toast_rel;
	Relation	fc_class_rel;
	Oid			fc_toast_relid;
	Oid			fc_namespaceid;
	char		fc_toast_relname[NAMEDATALEN];
	char		fc_toast_idxname[NAMEDATALEN];
	IndexInfo  *fc_indexInfo;
	Oid			fc_collationObjectId[2];
	Oid			fc_classObjectId[2];
	int16		fc_coloptions[2];
	ObjectAddress fc_baseobject,
				fc_toastobject;

	/*
	 * 是否已经被变成 toast？
	 */
	if (fc_rel->rd_rel->reltoastrelid != InvalidOid)
		return false;

	/*
	 * 检查表是否实际上需要一个 TOAST 表。
	 */
	if (!IsBinaryUpgrade)
	{
		/* 正常模式，正常检查 */
		if (!fc_needs_toast_table(fc_rel))
			return false;
	}
	else
	{
		/*
		 * 在二进制升级模式下，仅当 pg_upgrade 告诉我们时
		 * 创建一个 TOAST 表（即，提供了一个 TOAST 表 OID）。
		 *
		 * 这表明旧集群对当前表有一个 TOAST 表。我们必须创建一个 TOAST 表来接收旧的
		 * TOAST 文件，即使表似乎不需要一个。
		 *
		 * 相反，如果旧集群没有 TOAST 表，我们
		 * 应该能够没有一个，即使新版本的
		 * needs_toast_table 规则建议我们应该有一个。我们之间存在很大差距
		 * 创建 TOAST 表的位置和真正避免故障所需的位置，因此在何时创建的启发式原则上的小跨版本差异不应该是问题。
		 * 如果我们还是尝试创建 TOAST 表，我们将面临
		 * 它可能占用一个 OID，这将与我们尚未见过的一些
		 * 旧集群表产生冲突。
		 */
		if (!OidIsValid(binary_upgrade_next_toast_pg_class_oid))
			return false;
	}

	/*
	 * 如果请求，检查锁模式是否足够。这是一个交叉检查，以防在早期代码中出现错误或冲突的决策。
	 */
	if (fc_check && fc_lockmode != AccessExclusiveLock)
		elog(ERROR, "AccessExclusiveLock required to add toast table.");

	/*
	 * 创建 toast 表及其索引
	 */
	snprintf(fc_toast_relname, sizeof(fc_toast_relname),
			 "pg_toast_%u", fc_relOid);
	snprintf(fc_toast_idxname, sizeof(fc_toast_idxname),
			 "pg_toast_%u_index", fc_relOid);

	/* 这非常痛苦... 需要一个元组描述符 */
	fc_tupdesc = CreateTemplateTupleDesc(3);
	TupleDescInitEntry(fc_tupdesc, (AttrNumber) 1,
					   "chunk_id",
					   OIDOID,
					   -1, 0);
	TupleDescInitEntry(fc_tupdesc, (AttrNumber) 2,
					   "chunk_seq",
					   INT4OID,
					   -1, 0);
	TupleDescInitEntry(fc_tupdesc, (AttrNumber) 3,
					   "chunk_data",
					   BYTEAOID,
					   -1, 0);

	/*
	 * 确保 toast 表本身不会被变成 toast，否则我们将会
	 * 被“烤熟” :-(。这对于 chunk_data 是至关重要的，因为类型 bytea 是
	 * 可烤的；再检查其他两个，以确保。
	 */
	TupleDescAttr(fc_tupdesc, 0)->attstorage = TYPSTORAGE_PLAIN;
	TupleDescAttr(fc_tupdesc, 1)->attstorage = TYPSTORAGE_PLAIN;
	TupleDescAttr(fc_tupdesc, 2)->attstorage = TYPSTORAGE_PLAIN;

	/* Toast 字段不应被压缩 */
	TupleDescAttr(fc_tupdesc, 0)->attcompression = InvalidCompressionMethod;
	TupleDescAttr(fc_tupdesc, 1)->attcompression = InvalidCompressionMethod;
	TupleDescAttr(fc_tupdesc, 2)->attcompression = InvalidCompressionMethod;

	/*
	 * 常规关系的 toast 表放在 pg_toast 中；临时关系的则放在每个后端的临时 toast 表命名空间中。
	 */
	if (isTempOrTempToastNamespace(fc_rel->rd_rel->relnamespace))
		fc_namespaceid = GetTempToastNamespace();
	else
		fc_namespaceid = PG_TOAST_NAMESPACE;

	/* Toast 表仅在其父级时共享。 */
	fc_shared_relation = fc_rel->rd_rel->relisshared;

	/* 仅在其父级时映射 */
	fc_mapped_relation = RelationIsMapped(fc_rel);

	fc_toast_relid = heap_create_with_catalog(fc_toast_relname,
										   fc_namespaceid,
										   fc_rel->rd_rel->reltablespace,
										   fc_toastOid,
										   InvalidOid,
										   InvalidOid,
										   fc_rel->rd_rel->relowner,
										   table_relation_toast_am(fc_rel),
										   fc_tupdesc,
										   NIL,
										   RELKIND_TOASTVALUE,
										   fc_rel->rd_rel->relpersistence,
										   fc_shared_relation,
										   fc_mapped_relation,
										   ONCOMMIT_NOOP,
										   fc_reloptions,
										   false,
										   true,
										   true,
										   fc_OIDOldToast,
										   NULL);
	Assert(fc_toast_relid != InvalidOid);

	/* 使 toast 关系可见，否则 table_open 将失败 */
	CommandCounterIncrement();

	/* 这里实际上不需要 ShareLock，但还是采取它 */
	fc_toast_rel = table_open(fc_toast_relid, ShareLock);

	/*
	 * 在 chunk_id, chunk_seq 上创建唯一索引。
	 *
	 * 注意：正常的 TOAST 访问例程实际上可以仅在 chunk_id 上的单列索引下工作。然而，切片访问
	 * 例程同时使用这两个列，以便更快地访问单个 chunk。此外，我们需要它唯一，以检查重复 TOAST chunk OID 的可能性。这样，索引的效率也会更高，因为 btree 对大量相等的键不是很满意。
	 */

	fc_indexInfo = makeNode(IndexInfo);
	fc_indexInfo->ii_NumIndexAttrs = 2;
	fc_indexInfo->ii_NumIndexKeyAttrs = 2;
	fc_indexInfo->ii_IndexAttrNumbers[0] = 1;
	fc_indexInfo->ii_IndexAttrNumbers[1] = 2;
	fc_indexInfo->ii_Expressions = NIL;
	fc_indexInfo->ii_ExpressionsState = NIL;
	fc_indexInfo->ii_Predicate = NIL;
	fc_indexInfo->ii_PredicateState = NULL;
	fc_indexInfo->ii_ExclusionOps = NULL;
	fc_indexInfo->ii_ExclusionProcs = NULL;
	fc_indexInfo->ii_ExclusionStrats = NULL;
	fc_indexInfo->ii_OpclassOptions = NULL;
	fc_indexInfo->ii_Unique = true;
	fc_indexInfo->ii_NullsNotDistinct = false;
	fc_indexInfo->ii_ReadyForInserts = true;
	fc_indexInfo->ii_CheckedUnchanged = false;
	fc_indexInfo->ii_IndexUnchanged = false;
	fc_indexInfo->ii_Concurrent = false;
	fc_indexInfo->ii_BrokenHotChain = false;
	fc_indexInfo->ii_ParallelWorkers = 0;
	fc_indexInfo->ii_Am = BTREE_AM_OID;
	fc_indexInfo->ii_AmCache = NULL;
	fc_indexInfo->ii_Context = CurrentMemoryContext;

	fc_collationObjectId[0] = InvalidOid;
	fc_collationObjectId[1] = InvalidOid;

	fc_classObjectId[0] = OID_BTREE_OPS_OID;
	fc_classObjectId[1] = INT4_BTREE_OPS_OID;

	fc_coloptions[0] = 0;
	fc_coloptions[1] = 0;

	index_create(fc_toast_rel, fc_toast_idxname, fc_toastIndexOid, InvalidOid,
				 InvalidOid, InvalidOid,
				 fc_indexInfo,
				 list_make2("chunk_id", "chunk_seq"),
				 BTREE_AM_OID,
				 fc_rel->rd_rel->reltablespace,
				 fc_collationObjectId, fc_classObjectId, fc_coloptions, (Datum) 0,
				 INDEX_CREATE_IS_PRIMARY, 0, true, true, NULL);

	table_close(fc_toast_rel, NoLock);

	/*
	 * 将 toast 表的 OID 存储在父关系的 pg_class 行中
	 */
	fc_class_rel = table_open(RelationRelationId, RowExclusiveLock);

	if (!IsBootstrapProcessingMode())
	{
		/* 正常情况，采用事务更新 */
		fc_reltup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(fc_relOid));
		if (!HeapTupleIsValid(fc_reltup))
			elog(ERROR, "cache lookup failed for relation %u", fc_relOid);

		((Form_pg_class) GETSTRUCT(fc_reltup))->reltoastrelid = fc_toast_relid;

		CatalogTupleUpdate(fc_class_rel, &fc_reltup->t_self, fc_reltup);
	}
	else
	{
		/* 引导期间，我们无法 UPDATE，因此就地覆盖 */

		ScanKeyData fc_key[1];
		void	   *fc_state;

		ScanKeyInit(&fc_key[0],
					Anum_pg_class_oid,
					BTEqualStrategyNumber, F_OIDEQ,
					ObjectIdGetDatum(fc_relOid));
		systable_inplace_update_begin(fc_class_rel, ClassOidIndexId, true,
									  NULL, 1, fc_key, &fc_reltup, &fc_state);
		if (!HeapTupleIsValid(fc_reltup))
			elog(ERROR, "cache lookup failed for relation %u", fc_relOid);

		((Form_pg_class) GETSTRUCT(fc_reltup))->reltoastrelid = fc_toast_relid;

		systable_inplace_update_finish(fc_state, fc_reltup);
	}

	heap_freetuple(fc_reltup);

	table_close(fc_class_rel, RowExclusiveLock);

	/*
	 * 注册从 toast 表到主表的依赖关系，这样如果主表被删除，toast
	 * 表也将被删除。在引导模式下跳过此步骤。
	 */
	if (!IsBootstrapProcessingMode())
	{
		fc_baseobject.classId = RelationRelationId;
		fc_baseobject.objectId = fc_relOid;
		fc_baseobject.objectSubId = 0;
		fc_toastobject.classId = RelationRelationId;
		fc_toastobject.objectId = fc_toast_relid;
		fc_toastobject.objectSubId = 0;

		recordDependencyOn(&fc_toastobject, &fc_baseobject, DEPENDENCY_INTERNAL);
	}

	/*
	 * 使更改可见
	 */
	CommandCounterIncrement();

	return true;
}

/*
 * 检查表是否需要一个 TOAST 表。
 */
static bool fc_needs_toast_table(Relation fc_rel)
{
	/*
	 * 对于分区表，没必要创建 TOAST 表。
	 */
	if (fc_rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
		return false;

	/*
	 * 在 initdb 之后我们不能允许共享关系被变成 toast（因为
	 * 在其他数据库的 pg_class 中没有办法将其标记为 toasted）。
	 */
	if (fc_rel->rd_rel->relisshared && !IsBootstrapProcessingMode())
		return false;

	/*
	 * 在 initdb 之后忽略对目录表创建 toast 表的尝试。
	 * 哪些目录会有 toast 表是在 catalog/pg_*.h 中明确选择的。
	 * （如果目录没有 toast 表，我们可能会通过某些 ALTER TABLE 命令来到这里。）
	 */
	if (IsCatalogRelation(fc_rel) && !IsBootstrapProcessingMode())
		return false;

	/* 否则，让 AM 决定。 */
	return table_relation_needs_toast_table(fc_rel);
}
